package org.mushroomdb.lock;

import java.security.AccessController;
import java.security.Principal;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import javax.security.auth.Subject;

import org.mushroomdb.catalog.Catalog;
import org.mushroomdb.catalog.Column;
import org.mushroomdb.catalog.Register;
import org.mushroomdb.catalog.Table;
import org.mushroomdb.catalog.WriteBackRegister;
import org.mushroomdb.engine.EvaluationEngine;
import org.mushroomdb.engine.EvaluationEngineFactory;
import org.mushroomdb.engine.ResultSet;
import org.mushroomdb.exception.EvaluationEngineException;
import org.mushroomdb.exception.RDBMSRuntimeException;
import org.mushroomdb.filemanagement.Page;
import org.mushroomdb.lock.locktypes.Lock;
import org.mushroomdb.lock.locktypes.PageLock;
import org.mushroomdb.lock.locktypes.RegisterLock;
import org.mushroomdb.lock.locktypes.TableLock;
import org.mushroomdb.parser.ParserException;
import org.mushroomdb.query.Criteria;
import org.mushroomdb.query.impl.DeletionQuery;
import org.mushroomdb.query.impl.InsertionQuery;
import org.mushroomdb.query.impl.QueryHelper;
import org.mushroomdb.query.impl.SelectionQuery;
import org.mushroomdb.query.impl.UpdateQuery;
import org.mushroomdb.security.SecurityService;
import org.mushroomdb.security.User;
import org.mushroomdb.security.jaas.MushroomPrincipal;
import org.mushroomdb.service.Service;
import org.mushroomdb.service.ServiceManager;
import org.mushroomdb.transaction.TimeOutException;

/**
 * @author Tomas
 * 
 * TODO Refactorizar
 */
public class LockService implements Service, LockKeys {

    private boolean initialized = false;

    private LockConcurrencyManager lockConcurrencyManager;

    /**
     * Locks the register. If the register have already a Lock invokes the
     * LockConcurrencyManager, who is in charge of handle the conflict.
     * 
     * @param register
     * @return
     * @throws LockException
     */

    public Lock lock(Register register) throws LockException {
        SecurityService securityService = (SecurityService) ServiceManager
                .getInstance().getService(SecurityService.class);
        User user = securityService.getUserFromContext();

        Lock returnedLock = null;

        //TODO Refactorizar
        if (register instanceof WriteBackRegister) {

            Iterator iter = getLocks(register);
            if (iter == null) {
                throw new LockException();
            }

            //TODO Solo Locks exclusivos
            if (iter.hasNext()) {
                //el registro tiene un Lock
                RegisterLock lock = (RegisterLock) iter.next();
                if (lock.getUserId().equals(user.getName())) {
                    //el usuario en contexto ya tiene el Lock
                    return null;
                }

                try {
                    this.lockConcurrencyManager.append(lock, Thread
                            .currentThread());
                } catch (InterruptedException e1) {
                    throw new LockException(e1);
                }
            }

            try {
                //lockeo el registro
                returnedLock = lockRegister(register);
            } catch (LockException e) {
                throw new LockException("The register with Columns"
                        + register.getColumns().toString()
                        + " couldn't be Locked", e);
            }

        }
        return returnedLock;
    }

    /**
     * Locks the page. If the page is locked invokes the LockConcurrencyManager,
     * who is in charge of handle the conflict.
     * 
     * @param register
     * @return
     * @throws LockException
     */

    public Lock lock(Page page) throws LockException {

        //      TODO Refactorizar
        Lock returnedLock = null;

        Iterator iter = getLocks(page);
        if (iter == null) {
            throw new LockException(new RDBMSRuntimeException());
        }

        //TODO Solo Locks exclusivos
        if (iter.hasNext()) {
            //la pagina tiene un Lock
            PageLock lock = (PageLock) iter.next();
            try {
                this.lockConcurrencyManager
                        .append(lock, Thread.currentThread());
            } catch (InterruptedException e1) {
                throw new LockException(e1);

            }

        }

        try {
            returnedLock = lockPage(page);
        } catch (LockException e) {
            throw new LockException("The Page " + page.getLogicalNumber()
                    + " couldn't be Locked", e);
        }

        return returnedLock;
    }

    /**
     * Locks the table. If the table is locked invokes the
     * LockConcurrencyManager, who is in charge of handle the conflict.
     * 
     * @param register
     * @return
     * @throws LockException
     */

    public Lock lock(Table table) throws LockException {
        SecurityService securityService = (SecurityService) ServiceManager
                .getInstance().getService(SecurityService.class);
        User user = securityService.getUserFromContext();

        //      TODO Refactorizar
        Lock returnedLock = null;

        Iterator iter = getLocks(table);

        try {
            //TODO Solo Locks exclusivos
            if (iter.hasNext()) {
                //la tabla tiene un Lock
                TableLock lock = (TableLock) iter.next();

                if (lock.getUserId().equals(user.getName())) {
                    //el usuario en contexto ya tiene el Lock
                    return null;
                }
                this.lockConcurrencyManager
                        .append(lock, Thread.currentThread());
            }
        } catch (InterruptedException e) {
            throw new LockException(e);
        }

        try {
            returnedLock = lockTable(table);
        } catch (LockException e) {
            throw new LockException("The Table " + table.getName()
                    + " couldn't be Locked", e);
        }

        return returnedLock;
    }

    /**
     * Inserts a register in the SYS_LOCK table and into the SYS_LOCK_OBJECT
     * table
     * 
     * @param register
     * @return
     * @throws LockException
     */

    private synchronized Lock lockRegister(Register register)
            throws LockException {
        WriteBackRegister writeBackRegister = (WriteBackRegister) register;

        Catalog catalog = Catalog.getInstance();

        //Insert into LOCK_OBJECT_TABLE
        Table lockObjectTable = catalog.getTableByName(LOCK_OBJECT_TABLE);

        Long regId = new Long(QueryHelper.getInstance().getNewId(lockObjectTable
                .getColumn(LOCK_OBJECT_TABLE_ID)));

        Page page = writeBackRegister.getPage();
        Long pageNumber = new Long(page.getLogicalNumber());

        Long offset = new Long(writeBackRegister.getOffset());
        String delete = "N";

        //        SelectionQuery selectionQuery = new SelectionQuery();
        //        selectionQuery.addProjection(regId);
        //        selectionQuery.addProjection(pageNumber);
        //        selectionQuery.addProjection(offset);
        //
        //        InsertionQuery insertionQuery = new InsertionQuery(selectionQuery);
        //        insertionQuery.setTable(lockObjectTable);

        String query = "INSERT INTO SYS_LOCK_OBJECT (ID,PAGE,OFFSET,DELETE) VALUES ("
                + regId + "," + pageNumber + "," + offset + "," + delete + ")";
        InsertionQuery insertionQuery = null;
        try {
            insertionQuery = new InsertionQuery(query);
        } catch (ParserException e) {
            e.printStackTrace();
        }
        int updateCount = EvaluationEngineFactory.getInstance().executeUpdate(
                insertionQuery);

        if (updateCount != 1) {
            throw new LockException("The register " + register.toString()
                    + " couldn't be inserted in the " + LOCK_OBJECT_TABLE);
        }

        //Insert Into Lock_Table
        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Long lockId = new Long(QueryHelper.getInstance().getNewId(lockTable
                .getColumn(LOCK_TABLE_ID)));

        //the locked object it is a register
        //FIXME ver como hacer esto
        Long pageId = new Long(0);
        String tableId = null;

        Character exclusive = new Character('Y');

        SecurityService securityService = (SecurityService) ServiceManager
                .getInstance().getService(SecurityService.class);
        User user = securityService.getUserFromContext();

        //FIXME SACAR ESTO
        String userName = "DBO";
        if (user != null) {
            userName = user.getName();
        }

        Character type = new Character('R');

        //        SelectionQuery selectionQuery2 = new SelectionQuery();
        //        selectionQuery2.addProjection(lockId);
        //        selectionQuery2.addProjection(regId);
        //        selectionQuery2.addProjection(pageId);
        //        selectionQuery2.addProjection(tableId);
        //        selectionQuery2.addProjection(exclusive);
        //        selectionQuery2.addProjection(userName);
        //        selectionQuery2.addProjection(type);
        //
        //        InsertionQuery insertionQuery2 = new InsertionQuery(selectionQuery2);
        //        insertionQuery.setTable(lockTable);

        String query2 = "INSERT INTO SYS_LOCK (ID,REGISTER_ID,PAGE_ID,TABLE_ID,EXCLUSIVE,USER,TYPE,DELETE) VALUES ("
                + lockId
                + ","
                + regId
                + ","
                + pageId
                + ","
                + tableId
                + ","
                + exclusive + "," + userName + "," + type + "," + delete + ")";

        InsertionQuery insertionQuery2 = null;
        try {
            insertionQuery2 = new InsertionQuery(query2);
        } catch (ParserException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }
        int updateCount2 = EvaluationEngineFactory.getInstance().executeUpdate(
                insertionQuery2);

        if (updateCount2 != 1) {
            throw new LockException("The register " + register.toString()
                    + " couldn't be inserted in the " + LOCK_TABLE);
        }
        //FIXME Solo Modelo Lock\Unlock
        boolean ex = true;

        RegisterLock lock = new RegisterLock(lockId.longValue(), ex, userName,
                regId.longValue());

        return lock;
    }

    /**
     * Inserts a register in the SYS_LOCK table
     * 
     * @param table
     * @return
     * @throws LockException
     */

    private synchronized Lock lockTable(Table table) throws LockException {

        Catalog catalog = Catalog.getInstance();

        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Long lockId = new Long(QueryHelper.getInstance().getNewId(lockTable
                .getColumn(LOCK_TABLE_ID)));

        //the locked object it is a table

        //FIXME Sacar el ID de la Tabla mediante una query al motor
        //TODO:Ver como hacer esto
        Long pageId = new Long(0);
        Long regId = new Long(0);
        String tableId = table.getName();

        Character exclusive = new Character('Y');

        SecurityService securityService = (SecurityService) ServiceManager
                .getInstance().getService(SecurityService.class);
        User user = securityService.getUserFromContext();
        //FIXME:SACAR ESTO
        String userName = "DBO";
        if (user != null) {
            userName = user.getName();
        }

        Character type = new Character('T');
        String delete = "N";

        //        SelectionQuery selectionQuery = new SelectionQuery();
        //        selectionQuery.addProjection(lockId);
        //        selectionQuery.addProjection(regId);
        //        selectionQuery.addProjection(pageId);
        //        selectionQuery.addProjection(tableId);
        //        selectionQuery.addProjection(exclusive);
        //        selectionQuery.addProjection(userName);
        //        selectionQuery.addProjection(type);
        //
        //        InsertionQuery insertionQuery = new InsertionQuery(selectionQuery);
        //        insertionQuery.setTable(lockTable);
        //
        //        int updateCount =
        // EvaluationEngineFactory.getInstance().executeUpdate(
        //                insertionQuery);

        String query = "INSERT INTO SYS_LOCK (ID,REGISTER_ID,PAGE_ID,TABLE_ID,EXCLUSIVE,USER,TYPE,DELETE) VALUES ("
                + lockId
                + ","
                + regId
                + ","
                + pageId
                + ","
                + tableId
                + ","
                + exclusive + "," + userName + "," + type + "," + delete + ")";

        InsertionQuery insertionQuery = null;
        try {
            insertionQuery = new InsertionQuery(query);
        } catch (ParserException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }
        int updateCount = EvaluationEngineFactory.getInstance().executeUpdate(
                insertionQuery);

        if (updateCount != 1) {
            throw new LockException("The Table " + table.getName()
                    + " couldn't be inserted in the " + LOCK_TABLE);
        }
        //FIXME Solo Modelo Lock\Unlock
        boolean ex = true;

        TableLock lock = new TableLock(lockId.longValue(), ex, userName, table
                .getName());

        return lock;
    }

    /**
     * 
     * Inserts a register in the SYS_LOCK table
     * 
     * @param page
     * @return
     * @throws LockException
     */

    private synchronized Lock lockPage(Page page) throws LockException {
        Catalog catalog = Catalog.getInstance();

        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Long lockId = new Long(QueryHelper.getInstance().getNewId(lockTable
                .getColumn(LOCK_TABLE_ID)));

        //the locked object it is a page
        Long pageId = new Long(page.getLogicalNumber());
        Long regId = null;
        String tableId = null;

        Character exclusive = new Character('Y');

        SecurityService securityService = (SecurityService) ServiceManager
                .getInstance().getService(SecurityService.class);
        User user = securityService.getUserFromContext();
        String userName = user.getName();

        Character type = new Character('P');

        SelectionQuery selectionQuery = new SelectionQuery();
        selectionQuery.addProjection(lockId);
        selectionQuery.addProjection(regId);
        selectionQuery.addProjection(pageId);
        selectionQuery.addProjection(tableId);
        selectionQuery.addProjection(exclusive);
        selectionQuery.addProjection(userName);
        selectionQuery.addProjection(type);

        InsertionQuery insertionQuery = new InsertionQuery(selectionQuery,lockTable);

        int updateCount = EvaluationEngineFactory.getInstance().executeUpdate(
                insertionQuery);

        if (updateCount != 1) {
            throw new LockException("The Page " + page.getLogicalNumber()
                    + " couldn't be inserted in the " + LOCK_TABLE);
        }
        //FIXME Solo Modelo Lock\Unlock
        boolean ex = true;

        PageLock lock = new PageLock(lockId.longValue(), ex, userName, page
                .getLogicalNumber());

        return lock;
    }

    public void unlock(Lock lock) {
        if (lock.isPage()) {
            PageLock pageLock = (PageLock) lock;
            unlock(pageLock);
        } else if (lock.isRegister()) {
            RegisterLock registerLock = (RegisterLock) lock;
            unlock(registerLock);
        } else if (lock.isTable()) {
            TableLock tableLock = (TableLock) lock;
            unlock(tableLock);

        }

    }

    /**
     * Deletes the Lock from the tables SYS_LOCK_OBJECT and SYS_LOCK
     * 
     * @param pageLock
     */
    public void unlock(RegisterLock registerLock) {
        Catalog catalog = Catalog.getInstance();
        EvaluationEngine evaluationEngine = EvaluationEngineFactory
                .getInstance();
        Long reglockId = new Long(registerLock.getRegId());

        //borrar el lock de la tabla LOCK_OBJECT_TABLE
        /** SELECT * FROM SYS_LOCK_OBJECT WHERE ID= registerLock.id */
        Table lockObjectTable = catalog.getTableByName(LOCK_OBJECT_TABLE);

        Column lockObjectIdColumn = lockObjectTable
                .getColumn(LOCK_OBJECT_TABLE_ID);

        Column lockObjectDeleteColumn = lockObjectTable
                .getColumn(LOCK_OBJECT_TABLE_DELETE);

        SelectionQuery selectionQuery = new SelectionQuery();
        // selectionQuery.addProjection(lockObjectIdColumn);

        Criteria criteria = selectionQuery.getNewCriteria();
        criteria.addEqualsTo(lockObjectIdColumn, reglockId);

        UpdateQuery updateQuery = new UpdateQuery(selectionQuery);

        updateQuery.setUpdate(lockObjectDeleteColumn, "Y");

        int updateCount = EvaluationEngineFactory.getInstance().executeUpdate(
                updateQuery);
        if (updateCount != 1) {
            throw new RDBMSRuntimeException("El registro no pudo ser eliminado");
        }

        //FIXME:Descomentar cuando ande el delete
        //DeletionQuery deletionQuery2 = new DeletionQuery(selectionQuery2);

        //int count = evaluationEngine.executeUpdate(deletionQuery2);

        //Borrar el Lock de la LOCK_TABLE
        /** SELECT * FROM SYS_LOCK WHERE REGISTER_ID= registerLock.id */

        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
        Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
        Column lockDeleteColumn = lockTable.getColumn(LOCK_TABLE_DELETE);

        SelectionQuery selQuery = new SelectionQuery();
//        selQuery.addProjection(lockIdColumn);
//        selQuery.addProjection(lockRegIdColumn);
//        selQuery.addProjection(lockDeleteColumn);
       
        Criteria cri = selQuery.getNewCriteria();
        cri.addEqualsTo(lockRegIdColumn, reglockId);
        cri.addEqualsTo(lockDeleteColumn, "N");

        UpdateQuery upQuery2 = new UpdateQuery(selQuery);
        upQuery2.setUpdate(lockDeleteColumn, "Y");

        int updateCount2 = EvaluationEngineFactory.getInstance().executeUpdate(
                upQuery2);
        if (updateCount2 != 1) {
            throw new RDBMSRuntimeException(
                    "El Lock del registro no pudo ser eliminado");
        }
        
        //despertar el thread
        this.lockConcurrencyManager.remove(registerLock);

    }

    /**
     * Deletes the Lock from the table
     * 
     * @param tableLock
     */
    public void unlock(TableLock tableLock) {
        //borrar el lock de la tabla
        /** SELECT * FROM SYS_LOCK WHERE TABLE_ID= tableLock.id */

        Catalog catalog = Catalog.getInstance();
        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
        //        Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
        //        Column lockPageIdColumn = lockTable.getColumn(LOCK_TABLE_PAGE_ID);
        Column lockTableIdColumn = lockTable.getColumn(LOCK_TABLE_TABLE_ID);
        //        Column lockTypeColumn = lockTable.getColumn(LOCK_TABLE_TYPE);
        //        Column lockExclusiveColumn =
        // lockTable.getColumn(LOCK_TABLE_EXCLUSIVE);
        //        Column lockUserColumn = lockTable.getColumn(LOCK_TABLE_USER);
        Column lockDeleteColumn = lockTable.getColumn(LOCK_TABLE_DELETE);

        SelectionQuery selectionQuery = new SelectionQuery();
        //selectionQuery.addProjection(lockIdColumn);
        //selectionQuery.addProjection(lockRegIdColumn);
        //selectionQuery.addProjection(lockTableIdColumn);
        //        selectionQuery.addProjection(lockPageIdColumn);
        //        selectionQuery.addProjection(lockExclusiveColumn);
        //        selectionQuery.addProjection(lockUserColumn);
        //        selectionQuery.addProjection(lockTypeColumn);
        //selectionQuery.addProjection(lockDeleteColumn);

        //FIXME El ID NO debe ser el nombre de la tabla
        String tableId = tableLock.getTableId();

        Criteria criteria = selectionQuery.getNewCriteria();
        criteria.addEqualsTo(lockTableIdColumn, tableId);
        criteria.addEqualsTo(lockDeleteColumn, "N");

        EvaluationEngine evaluationEngine = EvaluationEngineFactory
                .getInstance();

        UpdateQuery updateQuery = new UpdateQuery(selectionQuery);
        updateQuery.setUpdate(lockDeleteColumn, "Y");

        int updateCount = evaluationEngine.executeUpdate(updateQuery);

        if (updateCount == 0) {
            throw new RDBMSRuntimeException(
                    "El Lock de la tabla no pudo ser eliminado");
        }

        //despertar el thread
        this.lockConcurrencyManager.remove(tableLock);
    }

    /**
     * Deletes the Lock from the table
     * 
     * @param pageLock
     */
    public void unlock(PageLock pageLock) {
        //borrar el lock de la tabla
        /** SELECT * FROM SYS_LOCK WHERE TABLE_ID= tableLock.id */

        Catalog catalog = Catalog.getInstance();
        Table lockTable = catalog.getTableByName(LOCK_TABLE);

        Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
        Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
        Column lockPageIdColumn = lockTable.getColumn(LOCK_TABLE_PAGE_ID);
        Column lockTableIdColumn = lockTable.getColumn(LOCK_TABLE_TABLE_ID);
        Column lockTypeColumn = lockTable.getColumn(LOCK_TABLE_TYPE);
        Column lockExclusiveColumn = lockTable.getColumn(LOCK_TABLE_EXCLUSIVE);
        Column lockUserColumn = lockTable.getColumn(LOCK_TABLE_USER);

        SelectionQuery selectionQuery = new SelectionQuery();
        selectionQuery.addProjection(lockIdColumn);
        selectionQuery.addProjection(lockRegIdColumn);
        selectionQuery.addProjection(lockTableIdColumn);
        selectionQuery.addProjection(lockPageIdColumn);
        selectionQuery.addProjection(lockExclusiveColumn);
        selectionQuery.addProjection(lockUserColumn);
        selectionQuery.addProjection(lockTypeColumn);

        Long pageId = new Long(pageLock.getId());

        Criteria criteria = selectionQuery.getNewCriteria();
        criteria.addEqualsTo(lockPageIdColumn, pageId);

        DeletionQuery deletionQuery = new DeletionQuery(selectionQuery);
        EvaluationEngine evaluationEngine = EvaluationEngineFactory
                .getInstance();

        int updateCount = evaluationEngine.executeUpdate(deletionQuery);

        //despertar el thread
        this.lockConcurrencyManager.remove(pageLock);
    }

    /**
     * Retrieves the Locks for the page passed. It makes a Selection Query on
     * the SYS_LOCK_OBJECT table to retrieve the ID for the Register passed.
     * With the Id, queries the SYS_LOCK table
     * 
     * @param register
     * @return
     */
    public Iterator getLocks(Register register) {
        if (this.initialized) {

            Catalog catalog = Catalog.getInstance();

            if (!(register instanceof WriteBackRegister)) {
                //            //TODO tirar una excepcion
                return null;
            }

            WriteBackRegister writeBackRegister = (WriteBackRegister) register;

            //hacer un query a la base
            /**
             * SELECT ID FROM SYS_LOCK_OBJECT WHERE PAGE= register.page AND
             * OFFSET= register.offset
             */

            Table lockObjectTable = catalog.getTableByName(LOCK_OBJECT_TABLE);

            Column lockObjectIdColumn = lockObjectTable
                    .getColumn(LOCK_OBJECT_TABLE_ID);
            Column lockObjectPageColumn = lockObjectTable
                    .getColumn(LOCK_OBJECT_TABLE_PAGE);
            Column lockObjectOffsetColumn = lockObjectTable
                    .getColumn(LOCK_OBJECT_TABLE_OFFSET);
            Column lockObjectDeleteColumn = lockObjectTable
                    .getColumn(LOCK_OBJECT_TABLE_DELETE);

            SelectionQuery selectionQuery = new SelectionQuery();
            selectionQuery.addProjection(lockObjectIdColumn);

            //TODO: VEr como hacer esto, Si page == null significa que el
            // registro
            // todavia
            // no fue persistido --> no tiene Locks--> devuelvo un iterador
            // vacio
            //        Page page = writeBackRegister.getPage();
            //        if(page==null){
            //            List list = new ArrayList();
            //            Iterator iter = list.iterator();
            //            return iter;
            //     }

            Page page = writeBackRegister.getPage();
            if (page == null) {
                //Se trata de un registro en trx, que no fue persistido
                //Ver bien como hacer esto
                List list = new ArrayList();
                Iterator iter = list.iterator();
                return iter;

            }
            Long pageNumber = new Long(page.getLogicalNumber());

            Criteria criteria = selectionQuery.getNewCriteria();

            criteria.addEqualsTo(lockObjectPageColumn, pageNumber);
            criteria.addEqualsTo(lockObjectOffsetColumn, new Long(
                    writeBackRegister.getOffset()));
            criteria.addEqualsTo(lockObjectDeleteColumn, "N");

            EvaluationEngine engine = EvaluationEngineFactory.getInstance();
            ResultSet resultSet = engine
                    .executeUnalterableQuery(selectionQuery);

            Long regId = null;
            try {
                if (resultSet.hasNext()) {
                    Register register2 = (Register) resultSet.next();
                    regId = (Long) register2.getValue(1);
                }
            } catch (EvaluationEngineException e) {
                throw new RDBMSRuntimeException(e);
            } finally {
            	resultSet.close();
            }

            if (regId == null) {
                //devuelvo un iterador vacio, pues el registro nunca se habia
                // lockeado
                //(no existe en la SYS_LOCK_OBJECT)
                List list = new ArrayList();
                Iterator iter = list.iterator();
                return iter;

            }

            /** SELECT * FROM SYS_LOCK WHERE REGISTER_ID= registerLock.id */
            Table lockTable = catalog.getTableByName(LOCK_TABLE);

            Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
            Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
            Column lockExclusiveColumn = lockTable
                    .getColumn(LOCK_TABLE_EXCLUSIVE);
            Column lockUserColumn = lockTable.getColumn(LOCK_TABLE_USER);
            Column lockDeleteColumn = lockTable.getColumn(LOCK_TABLE_DELETE);

            SelectionQuery selectionQuery2 = new SelectionQuery();
            selectionQuery2.addProjection(lockIdColumn);
            selectionQuery2.addProjection(lockRegIdColumn);
            selectionQuery2.addProjection(lockExclusiveColumn);
            selectionQuery2.addProjection(lockUserColumn);

            Criteria criteria2 = selectionQuery2.getNewCriteria();
            criteria2.addEqualsTo(lockRegIdColumn, regId);
            criteria2.addEqualsTo(lockDeleteColumn, "N");

            ResultSet resultSet2 = engine
                    .executeUnalterableQuery(selectionQuery2);

            List registerLockList = new ArrayList();

            try {
                while (resultSet2.hasNext()) {

                    Register reg = (Register) resultSet2.next();
                    Long lockId = (Long) reg.getValue(1);
                    Long registerId = (Long) reg.getValue(2);
                    String exclusive = (String) reg.getValue(3);
                    String userId = (String) reg.getValue(4);
                    boolean ex = ("Y").equalsIgnoreCase(exclusive);
                    RegisterLock registerLock = new RegisterLock(lockId
                            .longValue(), ex, userId, registerId.longValue());
                    registerLockList.add(registerLock);
                }

            } catch (EvaluationEngineException e1) {
                e1.printStackTrace();
                return null;
            }
            finally{
            	resultSet2.close();
            }
            return registerLockList.iterator();
        } else {
            //devuelvo un iterador vacio
            List list = new ArrayList();
            return list.iterator();
        }
    }

    /**
     * Retrieves the Locks for the table passed. It makes a Selection Query on
     * the SYS_LOCK table
     * 
     * @param table
     * @return
     */

    public Iterator getLocks(Table table) {
        if (this.initialized) {

            /** SELECT * FROM SYS_LOCK WHERE TABLE_ID= table.name */

            Catalog catalog = Catalog.getInstance();
            Table lockTable = catalog.getTableByName(LOCK_TABLE);

            Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
            Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
            Column lockPageIdColumn = lockTable.getColumn(LOCK_TABLE_PAGE_ID);
            Column lockTableIdColumn = lockTable.getColumn(LOCK_TABLE_TABLE_ID);
            Column lockTypeColumn = lockTable.getColumn(LOCK_TABLE_TYPE);
            Column lockExclusiveColumn = lockTable
                    .getColumn(LOCK_TABLE_EXCLUSIVE);
            Column lockUserColumn = lockTable.getColumn(LOCK_TABLE_USER);
            Column lockDeleteColumn = lockTable.getColumn(LOCK_TABLE_DELETE);

            SelectionQuery selectionQuery = new SelectionQuery();
            selectionQuery.addProjection(lockIdColumn);
            selectionQuery.addProjection(lockTableIdColumn);
            selectionQuery.addProjection(lockExclusiveColumn);
            selectionQuery.addProjection(lockUserColumn);
            selectionQuery.addProjection(lockTypeColumn);

            //FIXME El ID NO debe ser el nombre de la tabla
            String tableId = table.getName();

            Criteria criteria = selectionQuery.getNewCriteria();
            criteria.addEqualsTo(lockTableIdColumn, tableId);
            criteria.addEqualsTo(lockDeleteColumn, "N");

            EvaluationEngine engine = EvaluationEngineFactory.getInstance();

            ResultSet resultSet = engine
                    .executeUnalterableQuery(selectionQuery);

            List tableLockList = new ArrayList();

            try {
                while (resultSet.hasNext()) {
                    Register reg = (Register) resultSet.next();
                    Long lockId = (Long) reg.getValue(1);
                    String tabId = (String) reg.getValue(2);
                    String exclusive = (String) reg.getValue(3);
                    String userId = (String) reg.getValue(4);

                    boolean ex = ("Y").equalsIgnoreCase(exclusive);

                    TableLock tableLock = new TableLock(lockId.longValue(), ex,
                            userId, tabId);
                    tableLockList.add(tableLock);
                }
            } catch (EvaluationEngineException e1) {
                e1.printStackTrace();
                return null;
            }
            return tableLockList.iterator();
        } else {
            //devuelvo un iterador vacio
            List list = new ArrayList();
            return list.iterator();
        }
    }

    /**
     * Retrieves teh Locks for the page passed. It makes a Selection Query on
     * the SYS_LOCK table
     * 
     * @param page
     * @return
     */
    public Iterator getLocks(Page page) {
        if (this.initialized) {
            /** SELECT * FROM SYS_LOCK WHERE TABLE_ID= tableLock.id */

            Catalog catalog = Catalog.getInstance();
            Table lockTable = catalog.getTableByName(LOCK_TABLE);

            Column lockIdColumn = lockTable.getColumn(LOCK_TABLE_ID);
            Column lockRegIdColumn = lockTable.getColumn(LOCK_TABLE_REG_ID);
            Column lockPageIdColumn = lockTable.getColumn(LOCK_TABLE_PAGE_ID);
            Column lockTableIdColumn = lockTable.getColumn(LOCK_TABLE_TABLE_ID);
            Column lockTypeColumn = lockTable.getColumn(LOCK_TABLE_TYPE);
            Column lockExclusiveColumn = lockTable
                    .getColumn(LOCK_TABLE_EXCLUSIVE);
            Column lockUserColumn = lockTable.getColumn(LOCK_TABLE_USER);

            SelectionQuery selectionQuery = new SelectionQuery();
            selectionQuery.addProjection(lockIdColumn);
            selectionQuery.addProjection(lockPageIdColumn);
            selectionQuery.addProjection(lockExclusiveColumn);
            selectionQuery.addProjection(lockUserColumn);
            selectionQuery.addProjection(lockTypeColumn);

            Long pageId = new Long(page.getLogicalNumber());

            Criteria criteria = selectionQuery.getNewCriteria();
            criteria.addEqualsTo(lockPageIdColumn, pageId);

            EvaluationEngine engine = EvaluationEngineFactory.getInstance();

            List pageLockList = new ArrayList();

            ResultSet resultSet = engine
                    .executeUnalterableQuery(selectionQuery);

            try {
                while (resultSet.hasNext()) {
                    Register reg = (Register) resultSet.next();
                    Long lockId = (Long) reg.getValue(1);
                    Long pagId = (Long) reg.getValue(2);
                    String exclusive = (String) reg.getValue(3);
                    String userId = (String) reg.getValue(4);

                    boolean ex = ("Y").equalsIgnoreCase(exclusive);

                    PageLock pageLock = new PageLock(lockId.longValue(), ex,
                            userId, pagId.longValue());
                    pageLockList.add(pageLock);
                }
            } catch (EvaluationEngineException e1) {
                return null;
            }
            return pageLockList.iterator();
        } else {
            //devuelvo un iterador vacio
            List list = new ArrayList();
            return list.iterator();
        }
    }

    /**
     * Initialize the Lock Service and creates all the necessary tables to run
     * this services
     */
    public void initialize() {
        Catalog catalog = Catalog.getInstance();
        createLockTable(catalog);
        createLockObjectTable(catalog);

        this.initialized = true;
        this.lockConcurrencyManager = LockConcurrencyManager.getInstance();

    }

    /**
     * Creates the SYS_LOCK_OBJECT table, that holds the information about the
     * locked REGISTERS SYS_LOCK_OBJECT(ID (BIGINT), PAGE(INTEGER),
     * OFFSET(INTEGER), DELETE(INTEGER))
     */
    private void createLockObjectTable(Catalog catalog) {
        Table table = catalog.getTableByName(LOCK_OBJECT_TABLE);

        if (table == null) {

            Column id = new Column(LOCK_OBJECT_TABLE_ID, Types.BIGINT);
            //FIXME descomentar cuando ande el delete
            //id.setPk(true);

            Column page = new Column(LOCK_OBJECT_TABLE_PAGE, Types.BIGINT);

            Column offset = new Column(LOCK_OBJECT_TABLE_OFFSET, Types.BIGINT);

            Column delete = new Column(LOCK_OBJECT_TABLE_DELETE, Types.CHAR, 1);

            table = catalog.getBlankTable();
            table.setName(LOCK_OBJECT_TABLE);
            table.addColumn(id);
            table.addColumn(page);
            table.addColumn(offset);
            table.addColumn(delete);

            catalog.createTable(table);

        }

    }

    /**
     * Creates the SYS_LOCK table, that indicates the current Locks. The
     * Lockable objects are Table, Page and Register
     * 
     * FIXME Hacer un modelo Normalizado SYS_LOCK(ID (BIGINT), REG_ID(BIGINT),
     * TABLE_ID(CHAR,20), PAGE_ID(BIGINT),EXCLUSIVE(CHAR,1), USER_ID (CHAR,20),
     * TYPE (CHAR,1), DELETE(INTEGER))
     */

    private void createLockTable(Catalog catalog) {
        Table table = catalog.getTableByName(LOCK_TABLE);

        if (table == null) {

            Column id = new Column(LOCK_TABLE_ID, Types.BIGINT);
            //id.setPk(true);

            Column regId = new Column(LOCK_TABLE_REG_ID, Types.BIGINT);
            regId.setFk(true);
            regId.isAllowNulls();

            Column tableId = new Column(LOCK_TABLE_TABLE_ID, Types.CHAR, 20);
            tableId.setFk(true);
            tableId.isAllowNulls();

            Column pageId = new Column(LOCK_TABLE_PAGE_ID, Types.BIGINT);
            pageId.setFk(true);
            pageId.isAllowNulls();

            Column exclusive = new Column(LOCK_TABLE_EXCLUSIVE, Types.CHAR, 1);

            Column userId = new Column(LOCK_TABLE_USER, Types.CHAR, 20);
            userId.setFk(true);

            Column type = new Column(LOCK_TABLE_TYPE, Types.CHAR, 1);

            Column delete = new Column(LOCK_TABLE_DELETE, Types.CHAR, 1);

            table = catalog.getBlankTable();
            table.setName(LOCK_TABLE);
            table.addColumn(id);
            table.addColumn(regId);
            table.addColumn(tableId);
            table.addColumn(pageId);
            table.addColumn(exclusive);
            table.addColumn(userId);
            table.addColumn(type);
            table.addColumn(delete);

            catalog.createTable(table);

        }
    }

}
